package com.jackstraw.jack_base.mq.rocketmq.producer.impl;

import cn.hutool.core.lang.UUID;
import com.jackstraw.jack_base.mq.rocketmq.bean.RocketMQRoutingTemplate;
import com.jackstraw.jack_base.mq.rocketmq.producer.BaseProducer;
import com.jackstraw.jack_base.mq.rocketmq.bean.RocketMqMessage;
import com.jackstraw.jack_base.mq.common.entity.MQSendResult;
import com.jackstraw.jack_base.util.ObjectUtil;
import com.jackstraw.jack_base.util.StringUtil;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.SendCallback;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.client.producer.TransactionSendResult;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.messaging.Message;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.stereotype.Component;

import static com.jackstraw.jack_base.mq.common.entity.MQSendResult.DEFAULT;

@Slf4j
@Component
public class BaseProducerImpl implements BaseProducer {

    private final SendCallback sendCallback = new SendCallback() {
        @Override
        public void onSuccess(SendResult sendResult) {
            log.info("顺序消息发送成功:{}", sendResult);
        }

        @Override
        public void onException(Throwable e) {
            log.info("顺序消息发送失败:{}", e);

        }
    };

    @Autowired
    private RocketMQRoutingTemplate rocketMQRoutingTemplate;


    @Override
    public RocketMQTemplate getRocketMQTemplate() {
        return rocketMQRoutingTemplate.determineTargetEntity().getRocketMQTemplate();
    }


    @Override
    public DefaultMQProducer getDefaultMQProducer() {
        return rocketMQRoutingTemplate.determineTargetEntity().getDefaultMQProducer();
    }

    public Long getTimeout() {
        return (long) this.getDefaultMQProducer().getSendMsgTimeout();
    }


    @Override
    public MQSendResult send(RocketMqMessage<String> rocketMqMessage) {
        RocketMQTemplate rocketMQTemplate = this.getRocketMQTemplate();
        rocketMQTemplate.send(rocketMqMessage.getDestination(), this.buildMessage(rocketMqMessage));
        return DEFAULT();
    }

    @Override
    public <T> MQSendResult sendAndConvert(RocketMqMessage<T> rocketMqMessage, Class<T> tClass) {
        RocketMQTemplate rocketMQTemplate = this.getRocketMQTemplate();
        rocketMQTemplate.convertAndSend(rocketMqMessage.getDestination(),
                this.buildMessage(rocketMqMessage, tClass),
                rocketMqMessage.getHeadersByObject(), rocketMqMessage.getPostProcessor());
        return DEFAULT();
    }

    @Override
    public <T> MQSendResult sendAsyncOrderly(RocketMqMessage<T> rocketMqMessage, Class<T> tClass) {
        if (ObjectUtil.ObjectUtilByLang3.isEmpty(rocketMqMessage.getSendCallback())) {
            rocketMqMessage.setSendCallback(sendCallback);
            log.warn("顺序消息回调函数为空，将使用默认的回调函数");
        }

        if (StringUtil.StringUtilsBylang3.isEmpty(rocketMqMessage.getHashKey())) {
            rocketMqMessage.setHashKey(UUID.fastUUID().toString());
            log.warn("顺序消息HashKey为空，生成自动HashKey");
        }
        RocketMQTemplate rocketMQTemplate = this.getRocketMQTemplate();
        rocketMQTemplate.asyncSendOrderly(rocketMqMessage.getDestination(),
                this.buildMessage(rocketMqMessage, tClass),
                rocketMqMessage.getHashKey(), rocketMqMessage.getSendCallback(), rocketMqMessage.getTimeout(this.getTimeout()));
        return DEFAULT();
    }

    @Override
    public <T> MQSendResult<SendResult> sendSync(RocketMqMessage<T> rocketMqMessage, Class<T> tClass) {
        RocketMQTemplate rocketMQTemplate = this.getRocketMQTemplate();
        SendResult sendResult = rocketMQTemplate.syncSend(rocketMqMessage.getDestination(),
                this.buildMessage(rocketMqMessage, tClass),
                rocketMqMessage.getTimeout(this.getTimeout()));
        return DEFAULT(SendResult.class).setT(sendResult);
    }

    @Override
    public <T> MQSendResult<SendResult> sendSyncCollection(RocketMqMessage<T> rocketMqMessage, Class<T> tClass) {
        RocketMQTemplate rocketMQTemplate = this.getRocketMQTemplate();
        SendResult sendResult = rocketMQTemplate.syncSend(rocketMqMessage.getDestination(),
                rocketMqMessage.getMessage(), rocketMqMessage.getTimeout(this.getTimeout()));
        return DEFAULT(SendResult.class).setT(sendResult);
    }

    @Override
    public <T> MQSendResult<SendResult> sendSyncDelayLevel(RocketMqMessage<T> rocketMqMessage, Class<T> tClass) {
        if (ObjectUtil.ObjectUtilByLang3.isEmpty(rocketMqMessage.getDelayMessageLevelEnum())) {
            throw new IllegalStateException("延时消息中未设置延时时长");
        }
        RocketMQTemplate rocketMQTemplate = this.getRocketMQTemplate();
        SendResult sendResult = rocketMQTemplate.syncSend(rocketMqMessage.getDestination(),
                this.buildMessage(rocketMqMessage, tClass),
                rocketMqMessage.getTimeout((long) this.getDefaultMQProducer().getSendMsgTimeout()),
                rocketMqMessage.getDelayMessageLevelEnum().getDelayLevel());
        return DEFAULT(SendResult.class).setT(sendResult);
    }

    @Override
    public <T> MQSendResult<TransactionSendResult> sendMessageInTransaction(RocketMqMessage<T> rocketMqMessage, Class<T> tClass) {
        if (!isHaveTransactionListener(rocketMqMessage)) {
            log.error("事务消息Header中未包含TransactionListener验证的BeanName，要发送的消息为：", rocketMqMessage);
            return MQSendResult.ERROR().setResultMessage("事务消息Header中未包含TransactionListener验证的BeanName");
        }
        RocketMQTemplate rocketMQTemplate = this.getRocketMQTemplate();
        TransactionSendResult transactionSendResult = rocketMQTemplate.sendMessageInTransaction(rocketMqMessage.getDestination(),
                this.buildMessage(rocketMqMessage, tClass),
                rocketMqMessage.getArg());

        return MQSendResult.DEFAULT(TransactionSendResult.class).setT(transactionSendResult);
    }

    @Override
    public <T> Message<T> buildMessage(RocketMqMessage<T> rocketMqMessage, Class<T> tClass) {
        MessageBuilder<T> tMessageBuilder = MessageBuilder
                .withPayload(rocketMqMessage.getMessage());
        rocketMqMessage.getHeaders().forEach((key, value) -> {
            tMessageBuilder.setHeaderIfAbsent(key, value);
        });
        return tMessageBuilder.build();
    }

    @Override
    public Message<String> buildMessage(RocketMqMessage<String> rocketMqMessage) {
        return this.buildMessage(rocketMqMessage, String.class);
    }


}
